package org.apache.spark.examples

import org.apache.spark.{SparkConf, SparkContext}
import java.io.File

object RDDActionOperations {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("RDD Action Operations")
      .setMaster("local[4]")
    val sc = new SparkContext(conf)

    // 创建示例RDD
    val numbersRDD = sc.parallelize(1 to 20, 4)
    val pairRDD = sc.parallelize(List(("a", 1), ("b", 2), ("a", 3), ("c", 4)), 2)

    println("=== 基础行动算子 ===")
    
    // 1. collect：收集所有元素到Driver端
    val collected = numbersRDD.collect()
    println("\ncollect结果: " + collected.mkString("[", ", ", "]"))

    // 2. count：统计元素总数
    val count = numbersRDD.count()
    println(s"\ncount结果: $count")

    // 3. first：获取第一个元素
    val first = numbersRDD.first()
    println(s"\nfirst结果: $first")

    // 4. take：获取前n个元素
    val take3 = numbersRDD.take(3)
    println("\ntake(3)结果: " + take3.mkString(", "))

    // 5. top：获取最大的n个元素
    val top5 = numbersRDD.top(5)
    println("\ntop5结果: " + top5.mkString(", "))

    // 6. takeOrdered：获取最小的n个元素（可以自定义排序）
    val takeOrdered5 = numbersRDD.takeOrdered(5)(Ordering[Int].reverse)
    println("\ntakeOrdered5(降序)结果: " + takeOrdered5.mkString(", "))

    println("\n=== 聚合行动算子 ===")
    
    // 7. reduce：聚合操作
    val sum = numbersRDD.reduce(_ + _)
    println(s"\nreduce求和结果: $sum")

    // 8. fold：带初始值的聚合
    val foldResult = numbersRDD.fold(100)(_ + _)  // 初始值会参与每个分区的计算
    println(s"\nfold结果(初始值100): $foldResult")

    // 9. aggregate：更灵活的聚合
    val (sumCount, total) = numbersRDD.aggregate((0, 0))(
      (acc, num) => (acc._1 + num, acc._2 + 1),  // 分区内聚合
      (acc1, acc2) => (acc1._1 + acc2._1, acc1._2 + acc2._2)  // 分区间聚合
    )
    println(f"\naggregate结果: 总和=${sumCount} 数量=$total 平均值=${sumCount.toDouble/total}%.2f")

    println("\n=== 键值对行动算子 ===")
    
    // 10. countByKey：统计每个key的数量
    val countByKey = pairRDD.countByKey()
    println("\ncountByKey结果: " + countByKey)

    // 11. collectAsMap：转换为Map
    val collectAsMap = pairRDD.collectAsMap()
    println("\ncollectAsMap结果: " + collectAsMap)

    // 12. lookup：查找指定key的值
    val lookupA = pairRDD.lookup("a")
    println("\nlookup('a')结果: " + lookupA)

    println("\n=== 文件操作行动算子 ===")
    
    // 13. saveAsTextFile：保存结果到文件
    val outputPath = new File(System.getProperty("java.io.tmpdir"), "spark_output").getAbsolutePath
    numbersRDD.saveAsTextFile(outputPath)
    println(s"\n文件已保存到: $outputPath")

    // 14. 读取保存的文件验证
    val readRDD = sc.textFile(outputPath)
    println("\n读取保存文件的前5条记录:")
    readRDD.take(5).foreach(println)

    println("\n=== 其他重要行动算子 ===")
    
    // 15. foreach：对每个元素执行操作（通常用于副作用）
    println("\nforeach输出:")
    numbersRDD.foreach(x => print(x + " "))
    println()

    // 16. histogram：生成直方图数据
    val buckets = numbersRDD.histogram(5)
    println("\nhistogram(5个区间)结果:")
    println("区间边界: " + buckets._1.mkString(", "))
    println("区间计数: " + buckets._2.mkString(", "))

    // 17. 统计信息
    val stats = numbersRDD.stats()
    println("\n统计信息:")
    println(f"总数: ${stats.count}%d")
    println(f"平均值: ${stats.mean}%.2f")
    println(f"最大值: ${stats.max}%.0f")
    println(f"最小值: ${stats.min}%.0f")
    println(f"方差: ${stats.variance}%.2f")

    // 暂停以便查看Spark UI
    println("\n程序将暂停5分钟，请在此期间查看Spark UI...")
    Thread.sleep(300000)

    sc.stop()
  }
} 